package org.filesort.impl;

import java.io.*;
import java.util.ArrayList;
import java.util.List;

import javax.annotation.concurrent.Immutable;

import com.google.common.io.Files;
import com.google.common.primitives.Ints;
import org.filesort.*;
import org.filesort.IFileContentProcessor.IBytesSource;
import org.filesort.IFileContentProcessor.IBytesTarget;
import org.filesort.helpers.Formats;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;

/**
 * fixme: Class FileSorterImpl is for porn
 *
 * @author cheremin
 * @since 10.02.12,  16:17
 */
public class FileSorterImpl implements IFileSorter {
    private static final Logger log = LoggerFactory.getLogger( FileSorterImpl.class );

    private final ISorterConfiguration configuration;

    public FileSorterImpl( final ISorterConfiguration configuration ) {
        checkArgument( configuration != null, "configuration can't be null" );
        this.configuration = configuration;
    }

    @Override
    public void sort( final File source,
                      final IFileContentProcessor config,
                      final File target ) throws IOException {
        final int dataElementSize = config.dataElementSize();
        checkState( dataElementSize >= 1, "config.dataElementSize(%s) must be >=1", dataElementSize );
        final long totalFileLength = source.length();

        checkArgument(
                totalFileLength % dataElementSize == 0,
                "File [%s] size %s % dataElementSize(%s) != 0",
                source.getAbsolutePath(),
                totalFileLength,
                dataElementSize
        );


        final int bufferSize = align( configuration.ioBufferSize(), dataElementSize );

        final int aprioriRunLength = configuration.runLength();
        checkState(
                aprioriRunLength >= dataElementSize,
                "config.runLength(%s) must be >= dataElementSize(%s)",
                aprioriRunLength, dataElementSize
        );
        final int actualRunLength = align(
                ( int ) Math.min(
                        aprioriRunLength,
                        source.length()
                ),
                dataElementSize
        );

        log.info( "Sorting [" + source.getAbsolutePath() + "] with actualRunLength=" + actualRunLength );

        final int fullRunsCount = Ints.checkedCast( totalFileLength / actualRunLength );
        final int lastRunLength = Ints.checkedCast( totalFileLength % actualRunLength );

        final List<IRun> runs = new ArrayList<IRun>( fullRunsCount + 1 );
        for ( int i = 0; i < fullRunsCount; i++ ) {
            final IRun run = new RunImpl(
                    actualRunLength * ( long ) i,
                    actualRunLength
            );
            runs.add( run );
        }
        if ( lastRunLength > 0 ) {
            final IRun run = new RunImpl(
                    actualRunLength * ( long ) fullRunsCount,
                    lastRunLength
            );
            runs.add( run );
        }

        final File sortedRunsFile = configuration.createTemporaryFile( totalFileLength );
        sortedRunsFile.deleteOnExit();
        log.debug( runs.size() + " runs -> sorting into " + sortedRunsFile.getAbsolutePath() );

        //split and sort
        sortPhrase( source, sortedRunsFile, config, runs, bufferSize );

        log.debug( sortedRunsFile.getAbsolutePath() + " with sorted runs" );
        //second phase: merge sorted runs
        final File result = mergePhrase( config, sortedRunsFile, runs, bufferSize );

        Files.move( result, target );
    }

    private void sortPhrase( final File sourceFile,
                             final File sortedRunsFile,
                             final IFileContentProcessor configuration,
                             final List<IRun> runs,
                             final int bufferSize ) throws IOException {
        final FileInputStream fis = new FileInputStream( sourceFile );
        try {
            final BufferedInputStream bis = new BufferedInputStream( fis, bufferSize );
            try {
                final LimitedBytesSource bytesSource = new LimitedBytesSource( bis );
                final FileOutputStream fos = new FileOutputStream( sortedRunsFile );
                try {
                    final BufferedOutputStream bos = new BufferedOutputStream( fos, bufferSize );
                    try {
                        final LimitedBytesTarget bytesTarget = new LimitedBytesTarget( bos );
                        for ( final IRun run : runs ) {
                            final int runLength = Ints.checkedCast( run.length() );
                            bytesSource.setup( runLength );
                            bytesTarget.setup( runLength );
                            configuration.sortRun( bytesSource, bytesTarget, runLength );
                        }
                    } finally {
                        bos.close();
                    }
                } finally {
                    fos.close();
                }
            } finally {
                bis.close();
            }
        } finally {
            fis.close();
        }
    }

    private File mergePhrase( final IFileContentProcessor config,
                              final File sortedRunsFile,
                              final List<IRun> runs,
                              final int bufferSize ) throws IOException {
        final File[] mergeFiles = new File[]{
                sortedRunsFile,
                configuration.createTemporaryFile( sortedRunsFile.length() )
//                File.createTempFile( "merged-runs", ".data" )
        };
        for ( final File mergeFile : mergeFiles ) {
            mergeFile.deleteOnExit();
        }

        final int filesToMergeInOnePass = configuration.runsMergeInOnePass();
        int targetFileIndex = 0;
        List<IRun> currentRuns = new ArrayList<IRun>( runs );
        while ( true ) {
            final File sourceFile = mergeFiles[targetFileIndex];
            final File targetFile = mergeFiles[1 - targetFileIndex];
            log.debug( "merging: [" + sourceFile + "]@" + currentRuns.size() + " -> [" + targetFile + "]" );
            final RandomAccessFile raf = new RandomAccessFile( sourceFile, "r" );
            try {
                final FileOutputStream fos = new FileOutputStream( targetFile );
                try {
                    final BufferedOutputStream bos = new BufferedOutputStream( fos, bufferSize );
                    try {
                        final LimitedBytesTarget bytesTarget = new LimitedBytesTarget( bos );
                        final List<IRun> onePassRuns = new ArrayList<IRun>( filesToMergeInOnePass );
                        final List<IRun> nextRuns = new ArrayList<IRun>();
                        for ( final IRun run : currentRuns ) {
                            onePassRuns.add( run );
                            if ( onePassRuns.size() == filesToMergeInOnePass ) {
                                final IRun mergedRun = mergeRuns( config, onePassRuns, raf, bytesTarget );
                                nextRuns.add( mergedRun );
                                onePassRuns.clear();
                            }
                        }
                        if ( !onePassRuns.isEmpty() ) {
                            final IRun mergedRun = mergeRuns( config, onePassRuns, raf, bytesTarget );
                            nextRuns.add( mergedRun );
                        }

                        if ( nextRuns.size() == 1 ) {
                            sourceFile.delete();
                            return targetFile;
                        }
                        currentRuns = nextRuns;
                    } finally {
                        bos.close();
                    }
                } finally {
                    fos.close();
                }
            } finally {
                raf.close();
            }

            targetFileIndex = 1 - targetFileIndex;
        }
    }

    private IRun mergeRuns( final IFileContentProcessor configuration,
                            final List<IRun> runsToMerge,
                            final RandomAccessFile sourceRaf,
                            final LimitedBytesTarget bytesTarget ) throws IOException {
        /*
        todo optimize
        if(runsToMerge.size()  == 1 ){
            final IRun run = runsToMerge.get( 0 );
            final RandomAccessBytesSource source = new RandomAccessBytesSource( raf );
            source.setup( run.offset(), run.length() );
            bytesTarget.setup( run.length() );


        }*/
        log.debug( "\tmerging " + runsToMerge.size() + " runs" );
        final RandomAccessBytesSource[] sources = new RandomAccessBytesSource[runsToMerge.size()];
        long maxRunLength = 0;
        long totalRunLength = 0;
        long startOffset = Long.MAX_VALUE;
        for ( int i = 0; i < sources.length; i++ ) {
            final IRun run = runsToMerge.get( i );
            final RandomAccessBytesSource source = new RandomAccessBytesSource( sourceRaf );
            source.setup( run );
            sources[i] = source;

            maxRunLength = Math.max( run.length(), maxRunLength );
            totalRunLength += run.length();
            startOffset = Math.min( run.offset(), startOffset );
        }

        bytesTarget.setup( totalRunLength );
        configuration.merge(
                sources,
                maxRunLength,
                bytesTarget
        );
        log.debug( "\tmerged " + runsToMerge.size() + " runs" );
        return new RunImpl( startOffset, totalRunLength );
    }

    private static int align( final int size,
                              final int blockSize ) {
        checkArgument( size >= blockSize, "size(%s) must be >= block(%s)", size, blockSize );
        return ( size / blockSize ) * blockSize;
    }


    static class LimitedBytesTarget implements IBytesTarget {
        private final OutputStream stream;
        private long limit = -1;
        private long bytesLeft = 0;

        LimitedBytesTarget( final OutputStream stream ) {
            checkArgument( stream != null, "stream can't be null" );

            this.stream = stream;
        }

        @Override
        public void put( final byte[] buffer,
                         final int length ) throws IOException {
            checkState( limit > 0, "Not initialized" );

            checkArgument( buffer != null, "buffer can't ne null" );
            checkArgument( buffer.length >= length, "buffer.length(%s) < length(%s)", buffer.length, length );
            checkArgument( bytesLeft >= length, "bytesLeft(%s) < length(%s)", bytesLeft, length );


            stream.write( buffer, 0, length );

            bytesLeft -= length;
        }


        public void setup( final long limit ) {
            checkArgument( limit > 0, "limit(%s) must be >0", limit );
            checkState( bytesLeft == 0, "Can't reset if bytesLeft(%s) != 0", bytesLeft );

            this.bytesLeft = limit;
            this.limit = limit;
        }

        public long bytesWritten() {
            return limit - bytesLeft;
        }

        @Override
        public String toString() {
            return String.format(
                    "%s[limit:%d, left:%d]",
                    getClass().getSimpleName(),
                    limit,
                    bytesLeft
            );
        }
    }


    static class LimitedBytesSource implements IBytesSource {
        private final InputStream stream;

        private long limit = -1;
        private long bytesLeft = 0;

        LimitedBytesSource( final InputStream stream ) {
            checkArgument( stream != null, "stream can't be null" );

            this.stream = stream;
        }

        @Override
        public int get( final byte[] buffer ) throws IOException {
            checkState( limit > 0, "Not initialized" );

            checkArgument( buffer != null, "buffer can't ne null" );

            if ( bytesLeft == 0 ) {
                return 0;
            }

            final int bytesToRead = ( int ) Math.min( buffer.length, bytesLeft );

            final int wasRead = stream.read( buffer, 0, bytesToRead );

            if ( wasRead == 0 && bytesToRead != 0 ) {
                throw new IllegalStateException( "Stream was finished before all expected bytes was read (" + bytesLeft + " b left)" );
            }

            bytesLeft -= wasRead;

            return wasRead;
        }

        public void setup( final long limit ) {
            checkArgument( limit > 0, "limit(%s) must be >0", limit );
            checkState( bytesLeft == 0, "Can't reset if bytesLeft(%s)!=0", bytesLeft );

            this.bytesLeft = limit;
            this.limit = limit;
        }

        @Override
        public String toString() {
            return String.format(
                    "%s[limit:%d, left:%d]",
                    getClass().getSimpleName(),
                    limit,
                    bytesLeft
            );
        }
    }

    static class RandomAccessBytesSource implements IBytesSource {
        private final RandomAccessFile stream;

        private long limit = -1;
        private long bytesLeft = 0;
        private long position = 0;

        RandomAccessBytesSource( final RandomAccessFile stream ) {
            checkArgument( stream != null, "stream can't be null" );

            this.stream = stream;
        }

        @Override
        public int get( final byte[] buffer ) throws IOException {
            checkState( limit > 0, "Not initialized" );

            checkArgument( buffer != null, "buffer can't ne null" );

            if ( bytesLeft == 0 ) {
                return 0;
            }

            final int bytesToRead = ( int ) Math.min( buffer.length, bytesLeft );

            stream.seek( position );
            final int wasRead = stream.read( buffer, 0, bytesToRead );

            if ( wasRead == 0 && bytesToRead != 0 ) {
                throw new IllegalStateException( "Stream was finished (pos=" + position + ") before all expected bytes was read (" + bytesLeft + " b left)" );
            }

            bytesLeft -= wasRead;
            position += wasRead;

            return wasRead;
        }

        public void setup( final IRun run ) {
            setup( run.offset(), run.length() );
        }

        public void setup( final long startingPosition,
                           final long limit ) {
            checkArgument( limit > 0, "limit(%s) must be >0", limit );
            checkArgument( startingPosition >= 0, "startingPosition(%s) must be >0", startingPosition );
            checkState( bytesLeft == 0, "Can't setup(...) if bytesLeft(%s)!=0", bytesLeft );

            this.bytesLeft = limit;
            this.limit = limit;
            this.position = startingPosition;
        }

        @Override
        public String toString() {
            return String.format(
                    "%s[limit:%d, left:%d, pos:%d]",
                    getClass().getSimpleName(),
                    limit,
                    bytesLeft,
                    position
            );
        }
    }


    public interface IRun {
        /** @return offset of this run in file, bytes */
        public long offset();

        /** @return length of this run, bytes */
        public long length();
    }

    @Immutable
    public static class RunImpl implements IRun {
        private final long offset;
        private final long length;

        public RunImpl( final long offset,
                        final long length ) {
            checkArgument( offset >= 0, "offset(%s) must be >=0", offset );
            checkArgument( length > 0, "length(%s) must be >0", length );
            this.offset = offset;
            this.length = length;
        }

        @Override
        public long offset() {
            return offset;
        }

        @Override
        public long length() {
            return length;
        }

        @Override
        public String toString() {
            final StringBuilder sb = new StringBuilder();
            sb.append( "RunImpl[" )
                    .append( offset )
                    .append( "; +" )
                    .append( length )
                    .append( ']' );
            return sb.toString();
        }

    }

    public static void main( final String[] args ) throws Exception {
        final File source = new File( args[0] );

        final int runLength = 1 << Integer.parseInt( args[1] );
        final int mergeFilesInPass = Integer.parseInt( args[2] );
        final int bufferSize = 1 << Integer.parseInt( args[3] );//1Mb


        System.out.printf(
                "File: [%s] %s; run %s, merge %d files in pass, %s buffer\n",
                source.getAbsolutePath(),
                Formats.humanizedSize( source.length() ),
                Formats.humanizedSize( runLength ),
                mergeFilesInPass,
                Formats.humanizedSize( bufferSize )
        );


        final LongsProcessor config = new LongsProcessor( bufferSize );

        final FileSorterImpl sorter = new FileSorterImpl(
                new SimpleSorterConfiguration(
                        runLength,
                        mergeFilesInPass
                )
        );

//        isSorted( source );

        final long startedAt = System.currentTimeMillis();
        final File result = new File( source.getParent(), source.getName() + ".sorted" );

        sorter.sort( source, config, result );

        final long finishedAt = System.currentTimeMillis();
        System.out.printf(
                "Sorting done in %d ms\n",
                ( finishedAt - startedAt )
        );
        log.info( "Result: [" + result + "], " + Formats.humanizedSize( result.length() ) );
        log.info( "isSorted(): " + isSorted( result ) );
    }

    private static boolean isSorted( final File file ) throws IOException {
        final FileInputStream fis = new FileInputStream( file );
        try {
            final BufferedInputStream bis = new BufferedInputStream( fis, 1 << 20 );
            try {
                final DataInputStream dis = new DataInputStream( bis );
                try {
                    long prev = Long.MIN_VALUE;
                    while ( true ) {
                        final long l = dis.readLong();
                        if ( prev > l ) {
                            throw new IllegalStateException( "prev(" + prev + ") > cur(" + l + ")" );
                        }
                        prev = l;
                    }
                } catch ( EOFException e ) {
                    return true;
                } finally {
                    dis.close();
                }
            } finally {
                bis.close();
            }
        } finally {
            fis.close();
        }
    }
}
